package com.atguigu.day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

//CEP  判断账号连续3次登录失败
//另一种写法
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<LoginEvent> loginStream = env
                .fromElements(
                        LoginEvent.of("userId-1", "1.1.1.1", "fail", 1000L),
                        LoginEvent.of("userId-1", "1.1.1.2", "fail", 2000L),
                        LoginEvent.of("userId-2", "1.1.1.3", "fail", 3000L),
                        LoginEvent.of("userId-2", "1.1.1.4", "fail", 4000L),
                        LoginEvent.of("userId-1", "1.1.1.5", "fail", 5000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<LoginEvent>forMonotonousTimestamps()
                                .withTimestampAssigner((SerializableTimestampAssigner<LoginEvent>) (element, recordTimestamp) -> element.timestamp)
                );
        Pattern<LoginEvent, LoginEvent> loginEventPattern = Pattern.<LoginEvent>begin("failed")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return value.type.equals("fail");
                    }
                })
                .times(3)
                .consecutive();//连续
        PatternStream<LoginEvent> patternStream = CEP.pattern(loginStream.keyBy(x -> x.userId), loginEventPattern);
        patternStream.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                LoginEvent first = map.get("failed").get(0);
                LoginEvent second = map.get("failed").get(1);
                LoginEvent third = map.get("failed").get(2);
                return "账号: " + first.userId + " 连续3次登录失败  IP地址：" + first.ipAddress + " ; " + second.ipAddress + " ; " + third.ipAddress;
            }
        })
                .print();

        env.execute();
    }

    public static class LoginEvent {
        public String userId;
        public String ipAddress;
        public String type;
        public Long timestamp;

        public LoginEvent() {
        }

        public static LoginEvent of(String userId, String ipAddress, String type, Long timestamp) {
            return new LoginEvent(userId, ipAddress, type, timestamp);
        }

        private LoginEvent(String userId, String ipAddress, String type, Long timestamp) {
            this.userId = userId;
            this.ipAddress = ipAddress;
            this.type = type;
            this.timestamp = timestamp;
        }
    }
}
